# uncompyle6 version 3.2.3
# Python bytecode 3.6 (3379)
# Decompiled from: Python 3.6.8 |Anaconda custom (64-bit)| (default, Feb 21 2019, 18:30:04) [MSC v.1916 64 bit (AMD64)]
# Embedded file name: api\handlers\frontend\results.py
import json, logging, copy, uuid, re, os.path, pystache
from sqlalchemy import and_ as sql_and
from sqlalchemy import desc as sql_desc
from sqlalchemy import func as sql_func
from sqlalchemy import select as sql_select
from api.classes.frontend import *
from api.filters import Filters
from api.filters.sqlalchemy_adapter import get_sqlachemy_filter
from api.handlers.frontend import FrontendResourceHandler
from db import Connection
from db.data.access import *
from db.tables.locations import LocationRow, LocationsTable
from db.tables.scans import ScanRow, ScansTable
from db.tables.scan_session_vulns import ScanSessionVulnRow
from db.tables.scan_session_vulns_stats import (
    ScanSessionVulnsStatsTable,
    ScanSessionVulnsStatsRow,
)
from db.tables.vuln_types import VulnTypeRow, VulnTypesTable
from server.http_responses.http_status_codes import *
from server.http_responses.http_errors import Http404, Http409, ApiErrors
from db.tables.vuln_types import sensor_details_template
from db.data.users import require_at_least_tester, get_owner_id
from helpers.custom_profiles import generate_profile_for_vuln_recheck
from helpers.licensing.features import Features
from settings_provider import settings
from helpers.setup_types import *
from helpers.address import get_local_addresses
from helpers.excluded_hours import (
    is_allowed_to_scan_now,
    get_system_excluded_hours,
    get_next_date_allowed_to_scan,
)
from db.tables.targets_configurations import (
    TargetConfigurationTable,
    TargetConfigurationRow,
)
from server.http_responses.file_download import FileDownload

logger = logging.getLogger("api.Results")
root_table = LocationsTable.alias("root_table")
vulnerability_list_filters = {
    "keys": {
        "name": {
            "value_type": "string",
            "multiple_values": False,
            "operators": ("eq", "ilike_op"),
        },
        "severity": {
            "value_type": "int",
            "multiple_values": True,
            "operators": ("eq", "ne", "le", "ge", "lt", "gt"),
        },
        "status": {
            "value_type": "string",
            "allowed_values": ("open", "fixed", "ignored", "false_positive"),
            "multiple_values": True,
            "operators": ("eq", "ne"),
        },
        "rediscovered": {"value_type": "boolean", "operators": ("eq", "ne")},
        "cvss_score": {
            "value_type": "float",
            "multiple_values": True,
            "operators": ("eq", "ne", "le", "ge", "lt", "gt"),
        },
        "vuln_id": {"value_type": "int", "multiple_values": True},
        "vt_id": {"value_type": "uuid4", "multiple_values": True},
        "loc_id": {"value_type": "integer"},
    }
}
crawl_data_filters = {}


class Results(FrontendResourceHandler):
    def get_statistics(self, scan_id, result_id, **_):
        """
        Permissions: any_role & (access_all | access_group)
        :param scan_id:
        :param result_id:
        :param _:
        :return:
        """
        select_from, tail_filters = generate_scan_session_query(self.request_user)
        tail_filters.append(ScanSessionRow.scan_session_id == result_id)
        tail_filters.append(ScanSessionRow.scan_id == scan_id)
        select_from = select_from.outerjoin(
            ScanSessionVulnsStatsTable,
            ScanSessionVulnsStatsRow.scan_session_id == ScanSessionRow.scan_session_id,
        )
        select_from = select_from.outerjoin(
            ScanSessionJobsTable,
            ScanSessionJobRow.scan_session_id == ScanSessionRow.scan_session_id,
        )
        scan_session_query = sql_select(
            (
                ScanSessionJobRow,
                ScanSessionRow.target_id,
                ScanSessionRow.status.label("session_status"),
                ScanSessionRow.event_level,
                ScanSessionVulnsStatsRow.vuln_stats,
            )
        ).select_from(select_from)
        scan_session_query = apply_filters(scan_session_query, tail_filters)
        result = dict(scanning_app=dict())
        with Connection(self.current_shard_name) as (db):
            job_entries = db.execute(scan_session_query).fetchall()
            session_status = None
            severity_counts = None
            if not job_entries:
                return Http404()
            for job in job_entries:
                session_status = job.session_status
                if job.vuln_stats:
                    severity_counts = job.vuln_stats
                if not job.status:
                    break
                stats = None
                if job.status in ("starting", "processing", "aborting"):
                    scan_status_cache = settings.get("scan_status_cache")
                    if scan_status_cache:
                        stats = scan_status_cache.get_entry(
                            self.current_shard_name,
                            job.scanning_app,
                            job.scan_session_id,
                            enable_db_lookup=False,
                        )
                        if stats:
                            stats = copy.deepcopy(stats)
                if not stats:
                    stats = job.stats
                if not stats:
                    stats = {}
                stats["status"] = job.status
                stats["abort_requested"] = job.abort_requested
                stats["start_date"] = job.start_date
                stats["end_date"] = job.end_date
                stats["end_deadline"] = job.end_deadline
                stats["event_level"] = job.event_level
                result["scanning_app"][job.scanning_app] = stats

            if not severity_counts:
                severity_counts = dict(high=None, medium=None, low=None, info=None)
            else:
                if isinstance(severity_counts, (tuple, list)):
                    if len(severity_counts) == 4:
                        severity_counts = dict(
                            high=severity_counts[3],
                            medium=severity_counts[2],
                            low=severity_counts[1],
                            info=severity_counts[0],
                        )
            result["status"] = session_status
            result["severity_counts"] = severity_counts
        return result

    def get_location_vulnerabilities(self, loc_id, l, scan_id, result_id, c=None, **_):
        """
        Shortcut method to get_scan_vulnerabilities filtered by loc_id
        Permissions: any_role & (access_all | access_group)
        :param loc_id:
        :param l:
        :param scan_id:
        :param result_id:
        :param c:
        :param _:
        :return:
        """
        return self.get_scan_vulnerabilities(
            l=l,
            scan_id=scan_id,
            result_id=result_id,
            q="loc_id:" + str(loc_id),
            c=c,
            **_
        )

    def get_location_details(self, scan_id, loc_id, result_id, **_):
        """
        Permissions: any_role & (access_all | access_group)
        :param scan_id:
        :param loc_id:
        :param result_id:
        :param _:
        :return: ClsCrawlLocationDetails
        """
        joined_tables = []
        select_from, tail_filters = generate_scan_session_locations_extended_query(
            self.request_user, joined_tables
        )
        tail_filters.append(ScanSessionRow.scan_id == scan_id)
        tail_filters.append(ScanSessionRow.scan_session_id == result_id)
        tail_filters.append(LocationExtendedViewRow.loc_id == loc_id)
        select_from = select_from.outerjoin(
            root_table,
            sql_and(
                root_table.c.loc_id == LocationExtendedViewRow.root_id,
                root_table.c.scan_session_id == LocationExtendedViewRow.scan_session_id,
            ),
        )
        locations_query = sql_select(
            (LocationExtendedViewRow, root_table.c.name.label("root_url"))
        ).select_from(select_from)
        locations_query = apply_filters(locations_query, tail_filters)
        with Connection(self.current_shard_name) as (db_connection):
            location = db_connection.execute(locations_query).fetchone()
            if not location:
                return Http404()
            if location.root_url:
                if location.root_url.endswith("/"):
                    url = location.root_url[:-1] + location.path
                else:
                    url = location.root_url + location.path
            else:
                url = location.path
            location = dict(location)
            location["url"] = url
            location["severity_counts"] = dict(
                info=location["vuln_info"],
                low=location["vuln_low"],
                medium=location["vuln_medium"],
                high=location["vuln_high"],
            )
            if location["vuln_high"]:
                threat = 3
            else:
                if location["vuln_medium"]:
                    threat = 2
                else:
                    if location["vuln_low"]:
                        threat = 1
                    else:
                        threat = 0
                    location["threat"] = threat
                    result = ClsCrawlLocationDetails(**dict(location))
                    return result

    def search_crawl_data(self, l, scan_id, result_id, q=None, c=None, **_):
        """
        Search crawl data of this scan. If no search query parameter is passed
        to this response it simply returns a redirection to crawl root.
        Permissions: any_role & (access_all | access_group)
        :param l:
        :param scan_id:
        :param result_id:
        :param q:
        :param c:
        :param _:
        :return: ClsCrawlLocationListResponse
        """
        if q is None:
            return Http302(
                location="/api/v1/scans/%(scan_id)s/results/%(result_id)s/crawldata/0/children"
                % (dict(scan_id=scan_id, result_id=result_id))
            )
        else:
            limit = min(l, 100) if l is not None else 100
            c = int(c) if c is not None else 0
            cursor = c
            joined_tables = []
            select_from, tail_filters = generate_scan_session_locations_query(
                self.request_user, joined_tables
            )
            tail_filters.append(ScanSessionRow.scan_id == scan_id)
            tail_filters.append(ScanSessionRow.scan_session_id == result_id)
            locations_query = sql_select(
                (LocationRow, ScanSessionRow.target_id)
            ).select_from(select_from)
            locations_query = apply_filters(locations_query, tail_filters)
            locations_query = locations_query.order_by(
                sql_desc(LocationRow.loc_type), LocationRow.path, LocationRow.loc_id
            )
            locations_query = locations_query.limit(limit).offset(cursor)
            response = ClsCrawlLocationListResponse(
                locations=[],
                pagination=ClsPagination(previous_cursor=cursor, next_cursor=None),
            )
            with Connection(self.current_shard_name) as (db_connection):
                excluded_paths = None
                for loc in db_connection.execute(locations_query):
                    if excluded_paths is None:
                        select_from, filters = generate_target_query(self.request_user)
                        select_from = select_from.outerjoin(
                            TargetConfigurationTable,
                            sql_and(
                                TargetConfigurationRow.target_id == TargetRow.target_id,
                                TargetConfigurationRow.name == "excluded_paths",
                            ),
                        )
                        xp_query = (
                            sql_select(
                                (TargetConfigurationRow.value.label("excluded_paths"),)
                            )
                            .select_from(select_from)
                            .where(TargetRow.target_id == loc.target_id)
                        )
                        xp_query = apply_filters(xp_query, filters)
                        target = db_connection.execute(xp_query).fetchone()
                        excluded_paths = []
                        if target:
                            if target.excluded_paths:
                                excluded_paths = target.excluded_paths
                    loc = dict(loc)
                    for excluded_path in excluded_paths:
                        try:
                            if re.match(excluded_path, loc["path"]):
                                loc["excluded"] = True
                                break
                        except Exception:
                            pass

                    response.locations.append(ClsCrawlLocation(**loc))

            response_length = len(response.locations)
            if response_length >= limit:
                response.pagination.next_cursor = cursor + response_length
            return response

    def get_location_children(self, l, scan_id, result_id, loc_id, c=None, **_):
        """
        Permissions: any_role & (access_all | access_group)
        :param l:
        :param scan_id:
        :param result_id:
        :param loc_id:
        :param c:
        :param _:
        :return: ClsCrawlLocationListResponse
        """
        limit = min(l, 100) if l is not None else 100
        c = int(c) if c is not None else 0
        cursor = c
        loc_id_filter = (
            LocationRow.parent_id == loc_id
            if loc_id
            else LocationRow.parent_id.is_(None)
        )
        joined_tables = []
        select_from, tail_filters = generate_scan_session_locations_query(
            self.request_user, joined_tables
        )
        tail_filters.append(loc_id_filter)
        tail_filters.append(ScanSessionRow.scan_id == scan_id)
        tail_filters.append(ScanSessionRow.scan_session_id == result_id)
        locations_query = sql_select(
            (LocationRow, ScanSessionRow.target_id)
        ).select_from(select_from)
        locations_query = apply_filters(locations_query, tail_filters)
        locations_query = locations_query.order_by(
            sql_desc(LocationRow.loc_type), LocationRow.path, LocationRow.loc_id
        )
        locations_query = locations_query.limit(limit).offset(cursor)
        response = ClsCrawlLocationListResponse(
            locations=[],
            pagination=ClsPagination(previous_cursor=cursor, next_cursor=None),
        )
        with Connection(self.current_shard_name) as (db_connection):
            excluded_paths = None
            for loc in db_connection.execute(locations_query).fetchall():
                if excluded_paths is None:
                    select_from, filters = generate_target_query(self.request_user)
                    select_from = select_from.outerjoin(
                        TargetConfigurationTable,
                        sql_and(
                            TargetConfigurationRow.target_id == TargetRow.target_id,
                            TargetConfigurationRow.name == "excluded_paths",
                        ),
                    )
                    xp_query = (
                        sql_select(
                            (TargetConfigurationRow.value.label("excluded_paths"),)
                        )
                        .select_from(select_from)
                        .where(TargetRow.target_id == loc.target_id)
                    )
                    xp_query = apply_filters(xp_query, filters)
                    target = db_connection.execute(xp_query).fetchone()
                    excluded_paths = []
                    if target:
                        if target.excluded_paths:
                            excluded_paths = target.excluded_paths
                loc = dict(loc)
                for excluded_path in excluded_paths:
                    try:
                        if re.match(excluded_path, loc["path"]):
                            loc["excluded"] = True
                            break
                    except Exception:
                        pass

                response.locations.append(ClsCrawlLocation(**loc))

        response_length = len(response.locations)
        if response_length >= limit:
            response.pagination.next_cursor = cursor + response_length
        return response

    def get_scan_vulnerability_detail(self, scan_id, result_id, vuln_id, **_):
        """
        Permissions: any_role & (access_all | access_group)
        :param scan_id:
        :param result_id:
        :param vuln_id:
        :param _:
        :return: ClsVulnerabilityDetails
        """
        vuln_id = int(vuln_id)
        select_from, tail_filters = generate_scan_session_vulns_query(self.request_user)
        select_from = select_from.join(
            VulnTypesTable, VulnTypeRow.vt_id == ScanSessionVulnRow.vt_id
        )
        select_from = select_from.join(
            TargetVulnsTable, TargetVulnRow.vuln_hash == ScanSessionVulnRow.vuln_hash
        )
        select_from = select_from.outerjoin(
            LocationsTable,
            sql_and(
                LocationRow.loc_id == ScanSessionVulnRow.loc_id,
                LocationRow.scan_session_id == ScanSessionVulnRow.scan_session_id,
            ),
        )
        select_from = select_from.outerjoin(
            root_table,
            sql_and(
                root_table.c.loc_id == LocationRow.root_id,
                root_table.c.scan_session_id == ScanSessionVulnRow.scan_session_id,
            ),
        )
        query = sql_select(
            (
                ScanSessionVulnRow.request,
                ScanSessionVulnRow.tags,
                ScanSessionVulnRow.source,
                VulnTypeRow.tags.label("vt_tags"),
                VulnTypeRow.name,
                ScanSessionVulnRow.vt_id,
                TargetRow.target_id,
                TargetRow.criticality,
                TargetRow.type.label("target_type"),
                ScanSessionVulnRow.loc_detail,
                ScanSessionVulnRow.vuln_id,
                ScanSessionVulnRow.loc_id,
                LocationRow.path,
                root_table.c.name.label("root_url"),
                ScanSessionVulnRow.details,
                ScanSessionVulnRow.details_type,
                ScanSessionVulnRow.sensor_details,
                VulnTypeRow.severity,
                VulnTypeRow.impact,
                VulnTypeRow.description,
                VulnTypeRow.long_description,
                VulnTypeRow.recommendation,
                VulnTypeRow.refs,
                VulnTypeRow.cvss2,
                VulnTypeRow.cvss3,
                VulnTypeRow.cvss_score,
                VulnTypeRow.details_template,
                VulnTypeRow.app_id,
                ScanSessionVulnRow.vuln_hash,
                TargetVulnRow.status,
                TargetVulnRow.rediscovered,
                TargetVulnRow.http_response,
            )
        ).select_from(select_from)
        for tail_filter in tail_filters:
            query = query.where(tail_filter)

        query = query.where(ScanSessionVulnRow.vuln_id == vuln_id)
        query = query.where(ScanSessionRow.scan_id == scan_id)
        query = query.where(ScanSessionRow.scan_session_id == result_id)
        with Connection(self.current_shard_name) as (db_connection):
            vuln = db_connection.execute(query).fetchone()
            if not vuln:
                return Http404()
            vuln_tags = set(vuln.tags) if vuln.tags else set()
            vuln_type_tags = set(vuln.vt_tags) if vuln.vt_tags else set()
            tags = vuln_tags.union(vuln_type_tags)
            references = []
            if vuln.refs:
                for ref in vuln.refs:
                    references.append(ClsLink(rel=ref[0], href=ref[1]))

            if vuln.root_url:
                if vuln.root_url.endswith("/"):
                    url = vuln.root_url[:-1] + vuln.path
                else:
                    url = vuln.root_url + vuln.path
            else:
                url = vuln.path
            if vuln.status == "open":
                if vuln.rediscovered:
                    status = "rediscovered"
                status = vuln.status
            details = vuln.details
            try:
                if vuln.details_type == "json":
                    details = pystache.render(
                        vuln.details_template, json.loads(details)
                    )
            except Exception as e:
                logger.exception("render_details failed with %s for %s", e, vuln)

            try:
                if vuln.sensor_details:
                    sensor_details = pystache.render(
                        sensor_details_template, json.loads(vuln.sensor_details)
                    )
                    if not details:
                        details = sensor_details
                    else:
                        details += sensor_details
            except Exception as e:
                logger.exception("render sensor details failed with %s for %s", e, vuln)

            response_info = False
            if is_setup_type_on_premise():
                base_storage = settings.get("base_storage")
                vs = vuln.vuln_hash.replace("-", "")
                filename = os.path.join(
                    base_storage, "http_resp", vs[0], vs[1], vs[2], vs
                )
                response_info = os.path.exists(filename)
            else:
                if is_setup_type_aws():
                    if vuln.http_response is not None:
                        from helpers.aws.s3 import generate_link

                        vs = (
                            vuln.target_id.replace("-", "")
                            + "/"
                            + vuln.vuln_hash.replace("-", "")
                        )
                        response_info = generate_link(
                            settings["storage.http_responses.s3.bucket"], vs, 600
                        )
            data = dict(
                request=vuln.request,
                tags=list(tags),
                loc_id=vuln.loc_id,
                vt_id=vuln.vt_id,
                target_id=vuln.target_id,
                criticality=vuln.criticality,
                affects_url=url,
                vt_name=vuln.name,
                severity=vuln.severity,
                affects_detail=vuln.loc_detail,
                vuln_id=str(vuln_id),
                impact=vuln.impact,
                description=vuln.description,
                long_description=vuln.long_description,
                recommendation=vuln.recommendation,
                references=references,
                cvss2=vuln.cvss2,
                cvss3=vuln.cvss3,
                cvss_score=vuln.cvss_score,
                status=status,
                details=details,
                source=vuln.source,
                response_info=response_info,
            )
        return ClsVulnerabilityDetails(**data)

    def get_vulnerability_http_response(self, scan_id, result_id, vuln_id, **_):
        vuln_id = int(vuln_id)
        joined_tables = []
        select_from, tail_filters = generate_scan_session_vulns_query(
            self.request_user, joined_tables
        )
        q = (
            sql_select((ScanSessionVulnRow.vuln_hash,))
            .select_from(select_from)
            .where(ScanSessionVulnRow.vuln_id == vuln_id)
            .where(ScanSessionRow.scan_id == scan_id)
            .where(ScanSessionVulnRow.scan_session_id == result_id)
        )
        vs = None
        with Connection(self.current_shard_name) as (db):
            r = db.execute(q).fetchone()
            if not r:
                raise Http404()
            vs = r.vuln_hash.replace("-", "")
        if is_setup_type_on_premise():
            base_storage = settings.get("base_storage")
            filename = os.path.join(base_storage, "http_resp", vs[0], vs[1], vs[2], vs)
            if not os.path.exists(filename):
                raise Http404()
            return FileDownload(
                filename,
                headers={"Content-Encoding": "gzip"},
                mime_type="text/plain",
                open_file_mode="rb",
            )
        if is_setup_type_aws():
            raise Http404()
        raise Http404()

    def get_scan_vulnerabilities(self, scan_id, result_id, q=None, c=None, l=None, **_):
        """
        Permissions: any_role & (access_all | access_group)
        :param l:
        :param scan_id:
        :param result_id:
        :param q:
        :param c:
        :param _:
        :return: ClsVulnerabilityListResponse
        """
        limit = min(l, 100) if l is not None else 100
        c = int(c) if c is not None else 0
        cursor = c
        filters = Filters(vulnerability_list_filters, q)
        joined_tables = []
        select_from, tail_filters = generate_scan_session_vulns_query(
            self.request_user, joined_tables
        )
        select_from = select_from.outerjoin(
            TargetVulnsTable, TargetVulnRow.vuln_hash == ScanSessionVulnRow.vuln_hash
        )
        select_from = select_from.join(
            VulnTypesTable, VulnTypeRow.vt_id == ScanSessionVulnRow.vt_id
        )
        select_from = select_from.outerjoin(
            LocationsTable,
            sql_and(
                LocationRow.loc_id == ScanSessionVulnRow.loc_id,
                LocationRow.scan_session_id == ScanSessionVulnRow.scan_session_id,
            ),
        )
        select_from = select_from.outerjoin(
            root_table,
            sql_and(
                root_table.c.loc_id == LocationRow.root_id,
                root_table.c.scan_session_id == ScanSessionVulnRow.scan_session_id,
            ),
        )
        query = sql_select(
            (
                ScanSessionVulnRow.tags,
                ScanSessionVulnRow.vt_id,
                TargetRow.target_id,
                ScanSessionVulnRow.vuln_id,
                ScanSessionVulnRow.loc_detail,
                ScanSessionVulnRow.loc_id,
                LocationRow.path,
                root_table.c.name.label("root_url"),
                TargetRow.criticality,
                VulnTypeRow.name,
                VulnTypeRow.tags.label("vt_tags"),
                VulnTypeRow.severity,
                TargetVulnRow.status,
                TargetVulnRow.rediscovered,
            )
        )
        for filter_key, filter_expression in filters.data.items():
            flt = None
            if filter_key == "name":
                flt = get_sqlachemy_filter(
                    filter_expression,
                    VulnTypeRow.name,
                    filters.descriptor["keys"][filter_key],
                )
            else:
                if filter_key == "severity":
                    flt = get_sqlachemy_filter(
                        filter_expression,
                        VulnTypeRow.severity,
                        filters.descriptor["keys"][filter_key],
                    )
                else:
                    if filter_key == "status":
                        flt = get_sqlachemy_filter(
                            filter_expression,
                            TargetVulnRow.status,
                            filters.descriptor["keys"][filter_key],
                        )
                    else:
                        if filter_key == "rediscovered":
                            flt = get_sqlachemy_filter(
                                filter_expression,
                                TargetVulnRow.rediscovered,
                                filters.descriptor["keys"][filter_key],
                            )
                        else:
                            if filter_key == "cvss_score":
                                flt = get_sqlachemy_filter(
                                    filter_expression,
                                    VulnTypeRow.cvss_score,
                                    filters.descriptor["keys"][filter_key],
                                )
                            else:
                                if filter_key == "vt_id":
                                    flt = get_sqlachemy_filter(
                                        filter_expression,
                                        VulnTypeRow.vt_id,
                                        filters.descriptor["keys"][filter_key],
                                    )
                                else:
                                    if filter_key == "loc_id":
                                        flt = get_sqlachemy_filter(
                                            filter_expression,
                                            ScanSessionVulnRow.loc_id,
                                            filters.descriptor["keys"][filter_key],
                                        )
                                    else:
                                        if filter_key == "vuln_id":
                                            flt = get_sqlachemy_filter(
                                                filter_expression,
                                                ScanSessionVulnRow.vuln_id,
                                                filters.descriptor["keys"][filter_key],
                                            )
                                        else:
                                            if filter_key == "group_id":
                                                flt = get_sqlachemy_filter(
                                                    filter_expression,
                                                    Target2GroupRow.group_id,
                                                    filters.descriptor["keys"][
                                                        filter_key
                                                    ],
                                                )
                                                sf, tf = generate_target2group_query(
                                                    self.request_user
                                                )
                                                sq = sql_select(
                                                    (Target2GroupRow.target_id,)
                                                ).select_from(sf)
                                                sq = apply_filters(sq, tf)
                                                flt = ScanSessionRow.target_id.in_(
                                                    sq.where(flt)
                                                )
            if flt is not None:
                tail_filters.append(flt)

        query = query.select_from(select_from)
        for flt in tail_filters:
            query = query.where(flt)

        query = query.where(ScanSessionRow.scan_id == scan_id)
        query = query.where(ScanSessionRow.scan_session_id == result_id)
        query = query.order_by(
            sql_desc(VulnTypeRow.severity), VulnTypeRow.name, ScanSessionVulnRow.vuln_id
        )
        query = query.limit(limit).offset(cursor)
        result = ClsVulnerabilityListResponse(
            vulnerabilities=[],
            pagination=ClsPagination(previous_cursor=cursor, next_cursor=None),
        )
        with Connection(self.current_shard_name) as (db_connection):
            for vuln in db_connection.execute(query).fetchall():
                vuln_tags = set(vuln.tags) if vuln.tags else set()
                vuln_type_tags = set(vuln.vt_tags) if vuln.vt_tags else set()
                tags = vuln_tags.union(vuln_type_tags)
                if vuln.root_url:
                    if vuln.root_url.endswith("/"):
                        url = vuln.root_url[:-1] + vuln.path
                    else:
                        url = vuln.root_url + vuln.path
                else:
                    url = vuln.path
                if vuln.status == "open":
                    if vuln.rediscovered:
                        status = "rediscovered"
                    status = vuln.status
                vuln = dict(
                    last_seen=None,
                    tags=list(tags),
                    vuln_id=str(vuln.vuln_id),
                    target_id=vuln.target_id,
                    severity=vuln.severity,
                    affects_detail=vuln.loc_detail,
                    loc_id=vuln.loc_id,
                    status=status,
                    vt_id=vuln.vt_id,
                    affects_url=url,
                    vt_name=vuln.name,
                    criticality=vuln.criticality,
                )
                result.vulnerabilities.append(ClsVulnerability(**vuln))

        l = len(result.vulnerabilities)
        if l >= limit:
            result.pagination.next_cursor = cursor + l
        else:
            result.pagination.next_cursor = None
        return result

    def get_scan_vulnerability_types(
        self, result_id, scan_id, q=None, l=None, c=None, **_
    ):
        """
        Permissions: any_role & (access_all | access_group)
        :param result_id:
        :param scan_id:
        :param q:
        :param l:
        :param c:
        :param _:
        :return:
        """
        if not self.current_shard_name:
            raise AssertionError
        if not self.request_user:
            raise AssertionError
        limit = min(l, 100) if l is not None else 100
        c = int(c) if c is not None else 0
        cursor = c
        filters = Filters(vulnerability_list_filters, q)
        columns = (VulnTypesTable, sql_func.count(VulnTypeRow.vt_id).label("count"))
        tail_filters = [
            ScanSessionRow.scan_session_id == result_id,
            ScanSessionRow.scan_id == scan_id,
        ]
        if self.request_user["owner_id"] is None:
            from_clause = VulnTypesTable
            from_clause = from_clause.join(
                ScanSessionVulnsTable, ScanSessionVulnRow.vt_id == VulnTypeRow.vt_id
            )
            from_clause = from_clause.join(
                ScanSessionsTable,
                ScanSessionVulnRow.scan_session_id == ScanSessionRow.scan_session_id,
            )
            from_clause = from_clause.join(
                TargetsTable, TargetRow.target_id == ScanSessionRow.target_id
            )
            from_clause = from_clause.outerjoin(
                TargetVulnsTable,
                TargetVulnRow.vuln_hash == ScanSessionVulnRow.vuln_hash,
            )
            tail_filters.append(TargetRow.owner_id == self.request_user["user_id"])
        else:
            if self.request_user["access_all_groups"]:
                from_clause = VulnTypesTable
                from_clause = from_clause.join(
                    ScanSessionVulnsTable, ScanSessionVulnRow.vt_id == VulnTypeRow.vt_id
                )
                from_clause = from_clause.join(
                    ScanSessionsTable,
                    ScanSessionVulnRow.scan_session_id
                    == ScanSessionRow.scan_session_id,
                )
                from_clause = from_clause.join(
                    TargetsTable, TargetRow.target_id == ScanSessionRow.target_id
                )
                from_clause = from_clause.outerjoin(
                    TargetVulnsTable,
                    TargetVulnRow.vuln_hash == ScanSessionVulnRow.vuln_hash,
                )
                tail_filters.append(TargetRow.owner_id == self.request_user["owner_id"])
            else:
                from_clause = VulnTypesTable
                from_clause = from_clause.join(
                    ScanSessionVulnsTable, ScanSessionVulnRow.vt_id == VulnTypeRow.vt_id
                )
                from_clause = from_clause.join(
                    ScanSessionsTable,
                    ScanSessionVulnRow.scan_session_id
                    == ScanSessionRow.scan_session_id,
                )
                from_clause = from_clause.join(
                    TargetsTable, TargetRow.target_id == ScanSessionRow.target_id
                )
                from_clause = from_clause.outerjoin(
                    TargetVulnsTable,
                    TargetVulnRow.vuln_hash == ScanSessionVulnRow.vuln_hash,
                )
                tail_filters.append(
                    TargetRow.target_id.in_(
                        get_distinct_targets_id_by_group(self.request_user)
                    )
                )
            for filter_key, filter_expression in filters.data.items():
                flt = None
                if filter_key == "severity":
                    flt = get_sqlachemy_filter(
                        filter_expression,
                        VulnTypeRow.severity,
                        filters.descriptor["keys"][filter_key],
                    )
                else:
                    if filter_key == "status":
                        flt = get_sqlachemy_filter(
                            filter_expression,
                            TargetVulnRow.status,
                            filters.descriptor["keys"][filter_key],
                        )
                    else:
                        if filter_key == "rediscovered":
                            flt = get_sqlachemy_filter(
                                filter_expression,
                                TargetVulnRow.rediscovered,
                                filters.descriptor["keys"][filter_key],
                            )
                        else:
                            if filter_key == "cvss_score":
                                flt = get_sqlachemy_filter(
                                    filter_expression,
                                    VulnTypeRow.cvss_score,
                                    filters.descriptor["keys"][filter_key],
                                )
                            else:
                                if filter_key == "vt_id":
                                    flt = get_sqlachemy_filter(
                                        filter_expression,
                                        VulnTypeRow.vt_id,
                                        filters.descriptor["keys"][filter_key],
                                    )
                                else:
                                    if filter_key == "loc_id":
                                        flt = get_sqlachemy_filter(
                                            filter_expression,
                                            ScanSessionVulnRow.loc_id,
                                            filters.descriptor["keys"][filter_key],
                                        )
                                    else:
                                        if filter_key == "vuln_id":
                                            flt = get_sqlachemy_filter(
                                                filter_expression,
                                                ScanSessionVulnRow.vuln_id,
                                                filters.descriptor["keys"][filter_key],
                                            )
                if flt is not None:
                    tail_filters.append(flt)

            select = sql_select(columns).select_from(from_clause)
            for tail_filter in tail_filters:
                select = select.where(tail_filter)

            select = select.group_by(VulnTypeRow.vt_id)
            select = select.limit(limit).offset(cursor)
            select = select.order_by(
                sql_desc(VulnTypeRow.severity), VulnTypeRow.name, VulnTypeRow.vt_id
            )
            curred_vuln_types = []
            with Connection(self.current_shard_name) as (db_connection):
                for x in db_connection.execute(select).fetchall():
                    curred_vuln_types.append(
                        ClsVulnerabilityTypeSessionsCountResponseItem.from_dict(dict(x))
                    )

            l = len(curred_vuln_types)
            if l >= limit:
                next_cursor = cursor + l
            else:
                next_cursor = None
            return ClsVulnerabilityTypeSessionsCountResponse(
                vulnerability_types=curred_vuln_types,
                pagination=ClsPagination(
                    previous_cursor=cursor, next_cursor=next_cursor
                ),
            )

    def recheck_vulnerability(self, scan_id, result_id, vuln_id, body, **_):
        """
        :param body:
        :param scan_id:
        :param result_id:
        :param vuln_id:
        :param _:
        :return:
        """
        require_at_least_tester(self.request_user)
        self.license.require(Features.VULN_RETEST)
        vulnerability_recheck = ClsVulnerabilityRecheck(**body)
        joined_tables = []
        select_from, tail_filters = generate_scan_session_vulns_query(
            self.request_user, joined_tables
        )
        select_from = select_from.join(
            ScansTable, ScanRow.scan_id == ScanSessionRow.scan_id
        )
        select_from = select_from.join(
            VulnTypesTable, VulnTypeRow.vt_id == ScanSessionVulnRow.vt_id
        )
        select_from = select_from.join(
            LocationsTable,
            sql_and(
                LocationRow.loc_id == ScanSessionVulnRow.loc_id,
                LocationRow.scan_session_id == ScanSessionVulnRow.scan_session_id,
            ),
        )
        select_from = select_from.join(
            TargetsWithExcludedHoursViewTable,
            TargetsWithExcludedHoursViewRow.target_id == ScanRow.target_id,
        )
        query = sql_select(
            (
                TargetRow.target_id,
                TargetRow.address,
                TargetRow.description,
                TargetRow.manual_intervention,
                TargetsWithExcludedHoursViewRow.xh_data,
                TargetsWithExcludedHoursViewRow.xh_time_offset,
                VulnTypeRow.name,
                VulnTypeRow.app_id,
                ScanSessionVulnRow.source,
                ScanRow.profile_id,
            )
        ).select_from(select_from)
        query = (
            apply_filters(query, tail_filters)
            .where(ScanSessionRow.scan_id == scan_id)
            .where(ScanSessionRow.scan_session_id == result_id)
            .where(ScanSessionVulnRow.vuln_id == vuln_id)
        )
        with Connection(self.current_shard_name) as (db):
            vuln_info = db.execute(query).fetchone()
            if not vuln_info:
                raise Http404()
            system_excluded_hours = get_system_excluded_hours(db)
            target_xh_data = vuln_info.xh_data
            target_xh_time_offset = vuln_info.xh_time_offset
            ui_session_id = None
            if has_feature(SetupFeatures.MANUAL_BROWSING):
                if vuln_info.manual_intervention:
                    ui_session_id = vulnerability_recheck.get("ui_session_id")
                    if ui_session_id is None:
                        raise Http409(
                            "target requires manual intervention",
                            code=ApiErrors.MI_NO_SESSION_ID,
                        )
                    if self.remote_address not in get_local_addresses():
                        raise Http409(
                            "target requires manual intervention",
                            code=ApiErrors.MI_NOT_FROM_LOCALHOST,
                        )
                    if target_xh_data:
                        if not is_allowed_to_scan_now(
                            target_xh_data, target_xh_time_offset
                        ):
                            raise Http409(
                                "target requires manual intervention",
                                code=ApiErrors.MI_SCAN_NOW_NOT_POSSIBLE_TARGET_EXCLUDED_HOURS,
                            )
                        else:
                            if system_excluded_hours:
                                if not (
                                    is_allowed_to_scan_now(**system_excluded_hours)
                                ):
                                    raise Http409(
                                        "target requires manual intervention",
                                        code=ApiErrors.MI_SCAN_NOW_NOT_POSSIBLE_SYSTEM_EXCLUDED_HOURS,
                                    )
                        profile_id = generate_profile_for_vuln_recheck(
                            db=db,
                            source=vuln_info["source"],
                            creator_id=self.request_user["user_id"],
                            owner_id=get_owner_id(self.request_user),
                            original_profile_id=vuln_info["profile_id"],
                        )
                        if not profile_id:
                            raise Http409("vuln type not supported")
                        scan_id = str(uuid.uuid4())
                        scan_data = dict(
                            scan_id=scan_id,
                            owner_id=get_owner_id(self.request_user),
                            creator_id=self.request_user["user_id"],
                            target_id=vuln_info["target_id"],
                            profile_id=profile_id,
                            recurrence=None,
                            schedule_disabled=False,
                            history_limit=1,
                            schedule_start_date=None,
                            next_run=sql_func.now(),
                            schedule_time_sensitive=True,
                            report_template_id=None,
                        )
                        query_scan = ScansTable.insert().values(**scan_data)
                        db.execute(query_scan)
                        headers = None
                        if target_xh_data:
                            if not is_allowed_to_scan_now(
                                target_xh_data, target_xh_time_offset
                            ):
                                headers = {
                                    "X-Schedule-Delayed": str(
                                        get_next_date_allowed_to_scan(
                                            target_xh_data, target_xh_time_offset
                                        )
                                    )
                                }
                            else:
                                if system_excluded_hours:
                                    if not (
                                        is_allowed_to_scan_now(**system_excluded_hours)
                                    ):
                                        headers = {
                                            "X-Schedule-Delayed": str(
                                                get_next_date_allowed_to_scan(
                                                    **system_excluded_hours
                                                )
                                            )
                                        }
                            scan = dict(
                                scan_id=scan_id,
                                targetd=dict(
                                    target_id=vuln_info.target_id,
                                    address=vuln_info.address,
                                    description=vuln_info.description,
                                ),
                            )
                            if ui_session_id:
                                scan["ui_session_id"] = ui_session_id
                            if has_feature(SetupFeatures.MANUAL_BROWSING):
                                if vuln_info.manual_intervention:
                                    settings[
                                        "manual_browsing_storage"
                                    ].register_ui_session_id(scan_id, ui_session_id)
                            return Http201(
                                location=self.base_path
                                + "/scans/"
                                + scan_data["scan_id"],
                                body=scan,
                                headers=headers,
                            )
